package com.atguigu.userprofile.app;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.dao.TagInfoDAO;
import com.atguigu.userprofile.util.MyClickhouseUtil;
import com.atguigu.userprofile.util.MyPropertiesUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

public class TaskOutApp {
    public static void main(String[] args) {

        //获取业务日期
        String busiDate = args[1];

        //创建spark环境
        SparkConf sparkConf = new SparkConf().setAppName("TaskOutApp");//.setMaster("local[*]");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        //获取hive中画像库的名字
        Properties properties = MyPropertiesUtil.load("config.properties");
        String upDBname = properties.getProperty("user-profile.dbname");

        //获取Clickhouse中画像库的名字
        String ckDBname = properties.getProperty("clickhouse.dbname");
        //获取Clickhouse的url
        String ckUrl = properties.getProperty("clickhouse.url");

        //获取hive中表的名字
        String tableName = "up_tag_merge_" + busiDate.replace("-", "");

        //1.查询hive中的数据
        Dataset<Row> dataset = sparkSession.sql("select * from " + upDBname + "." + tableName);


        //2.创建Clickhouse中的宽表
        /**
         * create table if not exists user_profile.up_tag_merge_20200614
         *   (uid String, tg_person_base_gender String,tg_person_base_agegroup String)
         *   engine=MergeTree
         *   order by uid
         */
        //获取所有已启用标签的明细
        List<TagInfo> tagInfoListWithOn = TagInfoDAO.getTagInfoListWithOn();
        //获取所有已启用标签并拼接
        List<String> tagCodeWithOnList = tagInfoListWithOn.stream().map(tagInfo -> tagInfo.getTagCode().toLowerCase() + " String").collect(Collectors.toList());
        String tagCodeSQL = StringUtils.join(tagCodeWithOnList, ",");

        //为了保证重新跑的时候没有脏数据
        String dropSQL = "drop table if exists " + ckDBname + "." + tableName + ";";

        MyClickhouseUtil.executeSql(dropSQL);

        String createTable = " create table if not exists "+ckDBname+"."+tableName+" \n" +
                "  (uid String, "+tagCodeSQL+") \n" +
                "           engine=MergeTree \n" +
                "           order by uid";

        MyClickhouseUtil.executeSql(createTable);



        //3.将数据导入Clickhouse
        dataset.write().mode(SaveMode.Append)
                .option("driver", "ru.yandex.clickhouse.ClickHouseDriver")
                .option("batchsize", 500) //批量提交1.减少连接 网络IO次数 2.减少磁盘碎片
                .option("isolationLevel", "NONE")   //事务关闭
                .option("numPartitions", "4") // 设置并发
                .jdbc(ckUrl, tableName, new Properties());


    }
}
